package com.chaos.eurekaproducer.rabbitmq;

import com.alibaba.fastjson.JSON;
import com.chaos.eurekaproducer.domain.User;
import com.chaos.eurekaproducer.redis.RedisTemplateUtil;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.util.UUID;

/**
 * @program: wms
 * * @description:
 * * @author: liaopeng
 * * @create: 2020-04-23 16:07
 **/
@Component
public class RabbitMqProductor implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback{
    private Log log = LogFactory.getLog(getClass());

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisTemplateUtil jedisUtil;

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (!ack){
            log.info("发送mq消息失败:"+cause);
            String id = correlationData.getId();
            String value = jedisUtil.get(id).toString();
            String[] array = value.split("~");
            if (array.length==3){
                send(array[0],array[1],array[2]);//重发
            }
        }
    }

     /**
     * 发送mq消息
     * @param exchangeName 交换机
     * @param routingKey   路由键
     * @param msg       消息体
     */
    public void send(String exchangeName,String routingKey,String msg){
        try{
            String messageId = UUID.randomUUID().toString();//生成消息id
            log.info("messageId:"+messageId);
            CorrelationData correlationData = new CorrelationData(messageId);
            rabbitTemplate.setConfirmCallback(this);//消息确认回调
            rabbitTemplate.setMandatory(true);//交换机无法匹配到队列时返回false
            rabbitTemplate.setReturnCallback(this);//
            MessageProperties messageProperties = new MessageProperties();
            messageProperties.setMessageId(messageId);
            messageProperties.setContentType("text/plain");
            messageProperties.setContentEncoding("UTF-8");
            Message message = new Message(msg.getBytes(),messageProperties);
            rabbitTemplate.convertAndSend(exchangeName, routingKey, message,correlationData);
        }catch(Exception e){
            e.printStackTrace();
            log.error("convertAndSend消息异常,"+e.getMessage(),e);
        }
    }

    /**
     * 事务提交后再发消息
     * @param exchangeName
     * @param routingKey
     * @param msg
     */
    public void sendAfterTransaction(String exchangeName,String routingKey,String msg) {
        boolean actualTransactionActive = TransactionSynchronizationManager.isActualTransactionActive();
        log.info("当前消息是否持有事务："+actualTransactionActive);
        if (actualTransactionActive){
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
                @Override
                public void afterCommit() {
                    send(exchangeName,routingKey,msg);
                }
            });
        }else{
            send(exchangeName,routingKey,msg);
        }
    }

    /**
     * 匹配不到队列时的回调
     * @param message
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        String queue = message.getMessageProperties().getConsumerQueue();
        String messageId = message.getMessageProperties().getMessageId();
        String body = new String(message.getBody());
        log.info("匹配队列失败，exchange="+exchange+",queue="+queue+",routingKey="+routingKey+",body="+body);
        System.out.println("发送到队列失败");
    }
}
